﻿using Microsoft.Diagnostics.Tracing;
using Misakai.Kafka;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using ZooKeeperNet;
using Reflexive.IO.Kafka;

namespace Reflexive.IO
{
    public class KafkaReader<TConverter, TDestination> : KafkaBase, IReflexive<TConverter, byte[], TDestination>, IDisposable
        where TConverter : ITypeConverter<byte[], TDestination>, new()
        where TDestination : class
    {
        
        string _topic;
        string _endpoint;
        TopicOffset _offset;
        OffsetPosition[] _offsets;
        CancellationTokenSource _tokenSource;
        SynchronizedCollection<Subscription<TDestination>> _observers = new SynchronizedCollection<Subscription<TDestination>>();

        public StatusCode Status
        {
            get;
            private set;
        }

        public KafkaReader(IPAddress ip, int port, string topic, OffsetPosition[] offsets)
        {
            _topic = topic;
            _offsets = offsets;
            _offset = TopicOffset.Current;
            _endpoint = String.Format("{0}:{1}", ip.ToString(), port.ToString());
        }

        public KafkaReader(IPAddress ip, int port, string topic, TopicOffset offset = TopicOffset.End)
        {
            _topic = topic;
            _offset = offset == TopicOffset.Current ? TopicOffset.End : offset;
            _endpoint = String.Format("{0}:{1}",ip.ToString(), port.ToString());
        }

        public void Stop()
        {
            try
            {
                if (!_tokenSource.IsCancellationRequested)
                {
                    _tokenSource.Cancel();
                }
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.TraceError(e.ToString());
            }
            finally
            {
                Status = StatusCode.Suspended;
            }
        }

        public void Start()
        {
            if (Status == StatusCode.Completed)
            {
                throw new NotSupportedException("Reader is in the Completed state.");
            }

            _tokenSource = new CancellationTokenSource();
            
            Task.Run(() =>
            {

                var options = this.GetOptionsAsRetryable(_endpoint, _tokenSource.Token, new TimeSpan(0, 0, 0, 5)).Result; 
                KafkaConsumer consumer = new KafkaConsumer(new ConsumerOptions(_topic, new BrokerRouter(options)));

                switch (_offset)
                { 
                    case TopicOffset.Start:
                        var min = consumer.GetTopicOffsetSync(_topic).Select(result =>
                        {
                            return new OffsetPosition(result.PartitionId, result.Offsets.Min());
                        })
                        .OrderBy(offset => offset.PartitionId)
                        .ToArray();
                        consumer.SetOffsetPosition(min);
                        break;
                    case TopicOffset.End:
                         var max = consumer.GetTopicOffsetSync(_topic).Select(result => 
                        {
                            return new OffsetPosition(result.PartitionId, result.Offsets.Max());
                        })
                        .OrderBy(offset=>offset.PartitionId)
                        .ToArray();
                        consumer.SetOffsetPosition(max);
                        break;
                    default:
                        consumer.SetOffsetPosition(_offsets);
                        break;

                }

                try
                {
                    foreach (var data in consumer.Consume())
                    {
                        try
                        {
                            _observers.OnNext<TDestination>((Convertible<TConverter, byte[], TDestination>)data.Value);
                        }
                        catch (Exception e)
                        {
                            System.Diagnostics.Trace.TraceError(e.ToString());
                        }
                    }
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.TraceError(e.ToString());
                    try
                    {
                        _observers.OnError<TDestination>(e);
                    }
                    catch (Exception ex)
                    {
                        System.Diagnostics.Trace.TraceError(ex.ToString());
                    }
                    
                }

            }, _tokenSource.Token);
            Status = StatusCode.Started;
        }

        public void Dispose()
        {
            if (Status != StatusCode.Completed)
            {
                try
                {
                    if (!_tokenSource.IsCancellationRequested)
                    {
                        _tokenSource.Cancel();
                    }
                    _observers.OnCompleted<TDestination>();
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.TraceError(e.ToString());
                }
                finally
                {
                    Status = StatusCode.Completed;
                }
            }
        }

        #region Subscription Management

        public IDisposable Subscribe(IObserver<TDestination> observer)
        {
            var subscription = new Subscription<TDestination>(observer);
            subscription.UnSubscribe += UnSubscribe;
            _observers.Add(subscription);
            return subscription;
        }

        private void UnSubscribe(object sender, IObserver<TDestination> e)
        {
            _observers.Remove(((Subscription<TDestination>)sender));
        }

        #endregion

    }
}